package io.github.ali.commons.variable;

import java.util.concurrent.TimeUnit;

import org.apache.http.util.Asserts;

import io.github.ali.commons.function.NAFunction;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;

/**
 * 周期性触发动作</p>
 * 将由QuartzActionGenerator代替
 * @author fuli
 * @date 2018年11月6日
 * @version 1.0.0
 */
public class ActionGenerator extends ActionExecuter{
	private ObservableEmitter<Long> emitter;
	private Observable<Long> obva;
	private Disposable emitterDis;
	private long interval;
	private TimeUnit intervalTimeUnit;
	
	public ActionGenerator(long interval, TimeUnit timeUnit, NAFunction action) {
		super(action);
		Asserts.check(interval > 0, "interval must be positive!");
		Asserts.notNull(timeUnit, "timeUnit");
		this.interval = interval;
		this.intervalTimeUnit = timeUnit;
	}
	@Override
	protected void postStart(){
		this.obva = Observable.create((ObservableOnSubscribe<Long>) e -> {
			emitter = e;
			emitterDis = Observable.interval(interval, intervalTimeUnit).subscribe(e::onNext);
		});
		this.obva.subscribeOn(Schedulers.newThread()).observeOn(Schedulers.io()).subscribe((t)->triggerExecute(), (e) -> {
			LOGGER.error("the {}.{} execute error", this.getClass().getSimpleName(),name,e);
		}, () -> {
			LOGGER.info("the {}.{} has completed!",this.getClass().getSimpleName(),name);
		});
	}

	@Override
	protected void preStop() {
		emitterDis.dispose();
		emitter.onComplete();
		emitterDis = null;
		emitter = null;
		obva = null;
	}
}
